package source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Date 2021/6/16 10:18
 * 每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
 * 要求:
 * - 随机生成订单ID(UUID)
 * - 随机生成用户ID(0-2)
 * - 随机生成订单金额(0-100)
 * - 时间戳为当前系统时间
 *
 * SourceFunction:非并行数据源(并行度只能=1)
 * RichSourceFunction:多功能非并行数据源(并行度只能=1)
 * ParallelSourceFunction:并行数据源(并行度能够>=1)
 * RichParallelSourceFunction:多功能并行数据源(并行度能够>=1)--后续学习的Kafka数据源使用的就是该接口
 */
public class CustomSource01 {
    public static void main(String[] args) throws Exception {
        //1.env 创建 StreamExectution
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.source，创建自动生成 Order 数据源
        DataStreamSource<Order> source = env.addSource(new MyOrderSource());
        //3.打印数据源
        source.print();
        //4.执行
        env.execute();
        //定义实体类 Order 包括四个字段 oid uid money currentTime
        //定义静态内部类 MyOrderSource 继承 RichParallelSourceFunction
        //每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
        //要求:
        //- 随机生成订单ID(UUID)
        //- 随机生成用户ID(0-2)
        //- 随机生成订单金额(0-100)
        //- 时间戳为当前系统时间
    }

    public static class MyOrderSource extends RichParallelSourceFunction<Order> {
        boolean flag = true;
        Random rn = new Random();
        @Override
        public void run(SourceContext<Order> ctx) throws Exception {
            //每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)
            //要求:
            while(flag) {
                //- 随机生成订单ID(UUID)
                String oid = UUID.randomUUID().toString();
                //- 随机生成用户ID(0-2)
                int uid = rn.nextInt(3);
                //- 随机生成订单金额(0-100)
                int money = rn.nextInt(101);
                //- 时间戳为当前系统时间
                long currentTime = System.currentTimeMillis();
                ctx.collect(new Order(oid,uid,money,currentTime));
                //一秒钟休息一下
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
    //创建 Order 对象
    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class Order{
        private String oid;
        private int uid;
        private int money;
        private long currentTime;
    }
}
